package com.zhouqinuan.spring.cloud.develop.framework.rabbitmq.config;

import java.time.Duration;
import java.util.Date;
import java.util.stream.Collectors;

import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.autoconfigure.amqp.RabbitRetryTemplateCustomizer;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.zhouqinuan.spring.cloud.develop.framework.rabbitmq.RetryTemplateFactory;
import com.zhouqinuan.spring.cloud.develop.framework.rabbitmq.callback.DefaultConfirmCallback;
import com.zhouqinuan.spring.cloud.develop.framework.rabbitmq.callback.DefaultReturnCallback;
import com.zhouqinuan.spring.cloud.develop.framework.rabbitmq.handler.ListenerErrorHandler;
import com.zhouqinuan.spring.cloud.develop.framework.rabbitmq.handler.impl.DefaultListenerErrorHandler;

@Configuration
public class RabbitAutoConfiguration {

    private final RabbitProperties properties;

    private final ObjectProvider<RabbitRetryTemplateCustomizer> retryTemplateCustomizers;

    public RabbitAutoConfiguration(RabbitProperties properties,
            ObjectProvider<RabbitRetryTemplateCustomizer> retryTemplateCustomizers) {
        this.properties = properties;
        this.retryTemplateCustomizers = retryTemplateCustomizers;
    }
    
    @Bean
    public Jackson2JsonMessageConverter jackson2JsonMessageConverter() {
        Jackson2JsonMessageConverter messageConverter = new Jackson2JsonMessageConverter(new ObjectMapper());
        messageConverter.setCreateMessageIds(true);
        
        return messageConverter;
    }
    
    @Bean
    @ConditionalOnMissingBean
    public ListenerErrorHandler listenerErrorHandler() {
        
        return new DefaultListenerErrorHandler();
    }
    
    @Bean(name = "rabbitListenerContainerFactory")
    public SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory(
            SimpleRabbitListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory, ListenerErrorHandler listenerErrorHandler) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        configurer.configure(factory, connectionFactory);
        factory.setMessageConverter(jackson2JsonMessageConverter());
        factory.setErrorHandler(listenerErrorHandler);
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL); //开启手动ACK
        
        return factory;
    }
    
    @Bean
    @ConditionalOnMissingBean
    public ConfirmCallback confirmCallback() {
        
        return new DefaultConfirmCallback();
    }
    
    @Bean
    @ConditionalOnMissingBean
    public ReturnCallback returnCallback() {
        
        return new DefaultReturnCallback();
    }
    
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, ConfirmCallback confirmCallback, ReturnCallback returnCallback) {
        PropertyMapper map = PropertyMapper.get();
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        template.setMessageConverter(jackson2JsonMessageConverter());
        template.setMandatory(determineMandatoryFlag());
        RabbitProperties.Template properties = this.properties.getTemplate();
        if (properties.getRetry().isEnabled()) {
            template.setRetryTemplate(new RetryTemplateFactory(
                    this.retryTemplateCustomizers.orderedStream().collect(Collectors.toList())).createRetryTemplate(
                            properties.getRetry(), RabbitRetryTemplateCustomizer.Target.SENDER));
        }
        map.from(properties::getReceiveTimeout).whenNonNull().as(Duration::toMillis)
                .to(template::setReceiveTimeout);
        map.from(properties::getReplyTimeout).whenNonNull().as(Duration::toMillis).to(template::setReplyTimeout);
        map.from(properties::getExchange).to(template::setExchange);
        map.from(properties::getRoutingKey).to(template::setRoutingKey);
        map.from(properties::getDefaultReceiveQueue).whenNonNull().to(template::setDefaultReceiveQueue);
        
        template.setConfirmCallback(confirmCallback); //设置发送方确认回调
        template.setReturnCallback(returnCallback); //设置返回回调，消息无法发送到指定的队列的情况起作用
        
        template.setBeforePublishPostProcessors(new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {

                message.getMessageProperties().setTimestamp(new Date()); //默认发送消息设置时间戳，用于重试消费消息用，见方法AbstractMessageConsumer.processMessage

                return message;
            }
        });
        
        return template;
    }

    private boolean determineMandatoryFlag() {
        Boolean mandatory = this.properties.getTemplate().getMandatory();
        
        return (mandatory != null) ? mandatory : this.properties.isPublisherReturns();
    }
}
